package com.baishancloud.log.common.util

import cn.hutool.core.util.CharsetUtil
import cn.hutool.http.HttpUtil
import cn.hutool.json.{JSONArray, JSONObject, JSONUtil}
import com.baishancloud.log.common.entity._
import org.apache.commons.lang3.StringUtils
import org.apache.flink.util.ExceptionUtils
import org.slf4j.{Logger, LoggerFactory}

import java.io.InputStreamReader
import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, SQLException}
import java.util.Properties
import scala.collection.mutable
import scala.util.control.Breaks.{break, breakable}

class ExternalDataFetchUtil extends Serializable

/**
 * 获取外部缓存数据<br>
 * 每种资源的获取，都有两种方式：<br>
 * 　　一种是通过本地文件缓存获取，可用于算子的初始化，不受算子并行度的限制，没有高并发问题。<br>
 * 　　一种是直接去外部获取，比如查询 SR 表，或者是访问 HTTP ，可用于广播流，以获取到最新的数据。函数名称以 FromFile 结尾，使用时需注意。<br>
 * 另外，如果想要更新本地文件缓存的数据，需要去运行 test 中的 CacheDataToFile 代码，运行前请详细阅读相关注释。
 *
 * @author ziqiang.wang
 * @date 2022/02/16 11:28
 */
object ExternalDataFetchUtil extends Serializable {


  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[ExternalDataFetchUtil])


  //***************************************************************************starRocks连接相关操作***************************************************************************

  private val driverName: String = "com.mysql.jdbc.Driver"
  private val cjDriverName: String = "com.mysql.cj.jdbc.Driver"
  private val jdbcUrlPre: String = "jdbc:mysql://"
  private val jdbcUrlSuf: String = ":9030"
  private val ips = Set[String]("10.104.7.145", "10.104.7.41", "10.104.7.131", "10.104.7.146", "10.104.7.55")
  private val username: String = "portal"
  private val password: String = "portal@Baishancloud#1qaz"

  {
    //加载驱动类
    try {
      Class.forName(cjDriverName)
      LOGGER.info(s"成功加载驱动：$driverName")
    } catch {
      case _: ClassNotFoundException => {
        Class.forName(driverName)
        LOGGER.info(s"成功加载驱动：$cjDriverName")
      }
    }
  }

  /**
   * 获取连接
   */
  private def getConnection: Connection = {
    LOGGER.info("开始获取数据库连接")
    var connection: Connection = null
    breakable {
      for (ip <- ips) {
        LOGGER.info(s"正在通过${ip}来获取连接")
        connection = getConnectionFromIp(ip)
        if (connection != null) {
          LOGGER.info(s"通过${ip}获取连接成功")
          break()
        } else {
          LOGGER.info(s"通过${ip}获取连接失败")
        }
      }
    }
    if (connection == null) {
      throw new RuntimeException("喜鹊的 StarRocks 集群不可用，请检查集群状态")
    }
    connection
  }

  private def getConnectionFromIp(ip: String): Connection = {
    try {
      DriverManager.getConnection(jdbcUrlPre + ip + jdbcUrlSuf, username, password)
    } catch {
      case _: Exception => null
    }
  }


  /**
   * 关闭连接
   */
  private def closeConnection(connection: Connection): Unit = {
    if (connection == null) return
    try {
      connection.close()
    }
    catch {
      case e: SQLException => LOGGER.error("JDBC连接关闭失败", e)
    }
  }




  //***************************************************************************获取所有域名信息***************************************************************************

  /**
   * 获取域名信息对应的sql语句
   */
  private val domainInfoFetchSql: String = "select id as domain_id, name as domain_name, type as domain_type, user_id from cache_log.dim_domain order by id limit ? offset ?;"

  /**
   * 从starRocks表中查询域名信息，key 为域名名称。如果是用于算子初始化，建议使用[[domainInfoFetchFromFile]]。
   *
   * @param domainInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def domainInfoFetch(domainInfo: mutable.Map[String, DomainEntity]): Unit = {
    LOGGER.info("开始获取域名信息")
    val domainInfoInner: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(domainInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      domainInfoInner.clear()
      breakable(
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            domainInfoInner.put(rs.getString("domain_name"), DomainEntity(rs.getInt("domain_id"), rs.getString("domain_type"), rs.getInt("user_id")))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (domainInfoInner.isEmpty) {
      LOGGER.error(s"获取域名信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      domainInfo.clear()
      domainInfoInner.foreach(tuple => domainInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取域名信息成功，表数据量：${domainInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }

  /**
   * 通过文件缓存来获取域名信息，用于算子的初始化。key 为域名名称。
   *
   * @param domainInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def domainInfoFetchFromFile(domainInfo: mutable.Map[String, DomainEntity]): Unit = {
    LOGGER.info("开始从文件缓存中获取域名信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/domainInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: AnyRef, u: AnyRef) => {
      domainInfo.put(t.toString, JSONUtil.toBean(new String(u.toString), classOf[DomainEntity]))
    })
    LOGGER.info(s"从文件缓存中获取域名信息完成，总共获取到${domainInfo.size}条信息")
  }


  //***************************************************************************获取指定账户下的域名信息***************************************************************************

  // 获取指定账户下域名信息对应的sql语句
  private val accountDomainInfoFetchSqlPre: String = "select id as domain_id, name as domain_name, type as domain_type, user_id from cache_log.dim_domain where user_id in ( select id from cache_log.dim_user where name in ( "
  private val accountDomainInfoFetchMapSuf: String = " ) ) order by id limit ? offset ? ;"


  /**
   * 从starRocks表中查询指定账户的域名信息。key 为域名名称。如果是用于算子初始化，建议使用[[accountDomainInfoFetchFromFile]]。
   *
   * @param accounts          账户名称，使用英文逗号分隔
   * @param accountDomainInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def accountDomainInfoFetch(accounts: String, accountDomainInfo: mutable.Map[String, DomainEntity]): Unit = {
    LOGGER.info(s"开始获取指定账户下的域名信息，账户为：${accounts}")
    if (StringUtils.isEmpty(accounts)) {
      LOGGER.error("账户为空，直接返回")
      return
    }
    val accountDomainInfoInner: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
    var sqlMiddle: String = ""
    accounts.split(",").foreach(account => sqlMiddle += ("'" + account.trim + "'" + ", "))
    sqlMiddle = sqlMiddle.substring(0, sqlMiddle.length - 2)
    val accountDomainInfoFetchSql: String = accountDomainInfoFetchSqlPre + sqlMiddle + accountDomainInfoFetchMapSuf
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(accountDomainInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      accountDomainInfoInner.clear()
      breakable(
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            accountDomainInfoInner.put(rs.getString("domain_name"), DomainEntity(rs.getInt("domain_id"), rs.getString("domain_type"), rs.getInt("user_id")))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (accountDomainInfoInner.isEmpty) {
      LOGGER.error(s"获取指定账户下的域名信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      accountDomainInfo.clear()
      accountDomainInfoInner.foreach(tuple => accountDomainInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取指定账户下的域名信息成功，表数据量：${accountDomainInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }

  /**
   * 通过文件缓存来获取指定账户的域名信息，用于算子的初始化。key 为域名名称。
   *
   * @param accounts          账户名称，使用英文逗号分隔
   * @param accountDomainInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def accountDomainInfoFetchFromFile(accounts: String, accountDomainInfo: mutable.Map[String, DomainEntity]): Unit = {
    LOGGER.info(s"开始从文件缓存中获取指定账户下的域名信息，账户为：${accounts}")
    if (StringUtils.isEmpty(accounts)) {
      LOGGER.error("账户为空，直接返回")
      return
    }
    val accountDomainInfoInner: mutable.Map[String, AccountDomainEntity] = mutable.Map[String, AccountDomainEntity]()
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/accountDomainInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: scala.Any, u: scala.Any) => {
      accountDomainInfoInner.put(t.toString, JSONUtil.toBean(u.toString, classOf[AccountDomainEntity]))
    })
    val accountArray = accounts.split(",")
    for (elem: (String, AccountDomainEntity) <- accountDomainInfoInner) {
      if (accountArray.contains(elem._2.userName)) {
        accountDomainInfo.put(elem._1, DomainEntity(elem._2.domainId, elem._2.domainType, elem._2.userId))
      }
    }
    LOGGER.info(s"从文件缓存中获取指定账户下的域名信息完成，总共获取到${accountDomainInfo.size}条信息")
  }




  //***************************************************************************获取服务端对应cache组信息***************************************************************************

  /**
   * 获取服务端cache组信息对应的sql语句
   */
  private val serverCacheGroupInfoFetchSql: String = "select a.id as host_id, a.name as host_name, a.ip as host_ip, a.country as country_id, a.cache_group as cache_group_id, b.name as cache_group_name, b.code as cache_group_code from cache_log.dim_cache as a join cache_log.dim_cache_group as b on a.cache_group = b.id order by a.ip limit ? offset ?;"

  /**
   * 从starRocks表中查询服务端机器对应cache组信息。key 为机器的主机名。如果是用于算子初始化，建议使用[[serverCacheGroupFetchFromFile]]。
   *
   * @param serverCacheGroupInfo 外部缓存map表，函数会将结果写出到该map表中，key 为机器的主机名。
   */
  def serverCacheGroupFetch(serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity]): Unit = {
    LOGGER.info("开始获取服务端机器对应cache组信息")
    val ipCacheGroupInfoInner: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(serverCacheGroupInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      ipCacheGroupInfoInner.clear()
      breakable(
        //开始更新内部ipCacheGroupCodeInnerMap表，获取同步锁，防止读取到更新时的数据
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            ipCacheGroupInfoInner.put(rs.getString("host_name"), ServerCacheGroupEntity(rs.getInt("host_id"), rs.getString("host_ip"), rs.getString("host_name"), rs.getInt("country_id"), rs.getInt("cache_group_id"), rs.getString("cache_group_name"), rs.getString("cache_group_code")))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (ipCacheGroupInfoInner.isEmpty) {
      LOGGER.error(s"获取服务端机器对应cache组信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      serverCacheGroupInfo.clear()
      ipCacheGroupInfoInner.foreach(tuple => serverCacheGroupInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取服务端机器对应cache组信息成功，表数据量：${serverCacheGroupInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }

  /**
   * 通过文件缓存来获取服务端机器对应cache组信息，用于算子的初始化。key 为机器的主机名。
   *
   * @param serverCacheGroupInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def serverCacheGroupFetchFromFile(serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity]): Unit = {
    LOGGER.info("开始从文件缓存中获取服务端机器对应对应cache组信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/serverCacheGroupInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: AnyRef, u: AnyRef) => {
      serverCacheGroupInfo.put(t.toString, JSONUtil.toBean(u.toString, classOf[ServerCacheGroupEntity]))
    })
    LOGGER.info(s"从文件缓存中获取服务端机器对应对应cache组信息完成，总共获取到${serverCacheGroupInfo.size}条信息")
  }



  //***************************************************************************获取所有国家信息***************************************************************************

  /**
   * 获取所有国家信息对应的sql语句
   */
  private val countryInfoFetchSql: String = "select name, id, code from cache_log.dim_country order by id limit ? offset ?;"

  /**
   * 从starRocks中查询国家信息。key 为国家中文名称。如果是用于算子初始化，建议使用[[countryInfoFetchFromFile]]。
   *
   * @param countryInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def countryInfoFetch(countryInfo: mutable.Map[String, CountryEntity]): Unit = {
    LOGGER.info("开始更新countryIdInnerMap表")
    val countryInfoInner: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(countryInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      countryInfoInner.clear()
      breakable(
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            countryInfoInner.put(rs.getString("name"), CountryEntity(rs.getInt("id"), rs.getString("code")))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (countryInfoInner.isEmpty) {
      LOGGER.error(s"获取所有国家信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      countryInfo.clear()
      countryInfoInner.foreach(tuple => countryInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有国家信息成功，表数据量：${countryInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }


  /**
   * 通过文件缓存来获取国家信息，用于算子的初始化。key 为国家中文名。
   *
   * @param countryInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def countryInfoFetchFromFile(countryInfo: mutable.Map[String, CountryEntity]): Unit = {
    LOGGER.info("开始从文件缓存中获取国家信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/countryInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: AnyRef, u: AnyRef) => {
      countryInfo.put(t.toString, JSONUtil.toBean(u.toString, classOf[CountryEntity]))
    })
    LOGGER.info(s"从文件缓存中获取国家信息完成，总共获取到${countryInfo.size}条信息")
  }



  //***************************************************************************获取所有省份信息***************************************************************************

  /**
   * 获取所有省份信息对应的sql语句
   */
  private val provinceInfoFetchSql: String = "select name, id, code from cache_log.dim_province order by id limit ? offset ?;"

  /**
   * 从starRocks中查询省份信息。key 为省份中文名。如果是用于算子初始化，建议使用[[provinceInfoFetchFromFile]]。
   *
   * @param provinceInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def provinceInfoFetch(provinceInfo: mutable.Map[String, ProvinceEntity]): Unit = {
    LOGGER.info("开始获取所有省份信息")
    val provinceInfoInner: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(provinceInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      provinceInfoInner.clear()
      breakable(
        //开始更新内部provinceIdInnerMap表，获取同步锁，防止读取到更新时的数据
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            provinceInfoInner.put(rs.getString("name"), ProvinceEntity(rs.getInt("id"), rs.getString("code")))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (provinceInfoInner.isEmpty) {
      LOGGER.error(s"获取所有省份信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      provinceInfo.clear()
      provinceInfoInner.foreach(tuple => provinceInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有省份信息成功，表数据量：${provinceInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }


  /**
   * 通过文件缓存来获取省份信息，用于算子的初始化。key 为省份中文名。<br>
   *
   * @param provinceInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def provinceInfoFetchFromFile(provinceInfo: mutable.Map[String, ProvinceEntity]): Unit = {
    LOGGER.info("开始从文件缓存中获取省份信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/provinceInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: AnyRef, u: AnyRef) => {
      provinceInfo.put(t.toString, JSONUtil.toBean(u.toString, classOf[ProvinceEntity]))
    })
    LOGGER.info(s"从文件缓存中获取省份信息完成，总共获取到${provinceInfo.size}条信息")
  }



  //***************************************************************************获取所有运营商信息***************************************************************************

  /**
   * 获取所有运营商信息对应的sql语句
   */
  private val ispInfoFetchSql: String = "select name, id, code from cache_log.dim_isp order by id limit ? offset ?;"

  /**
   * 从starRocks中查询所有运营商信息。key 为运营商中文名。如果是用于算子初始化，建议使用[[ispInfoFetchFromFile]]。
   *
   * @param ispInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def ispInfoFetch(ispInfo: mutable.Map[String, IspEntity]): Unit = {
    LOGGER.info("开始获取所有运营商信息")
    val ispInfoInner: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(ispInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      ispInfoInner.clear()
      breakable(
        //开始更新内部ispIdInnerMap表，获取同步锁，防止读取到更新时的数据
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            ispInfoInner.put(rs.getString("name"), IspEntity(rs.getInt("id"), rs.getString("code")))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (ispInfoInner.isEmpty) {
      LOGGER.error(s"获取所有运营商信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      ispInfo.clear()
      ispInfoInner.foreach(tuple => ispInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有运营商信息成功，表数据量：${ispInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }


  /**
   * 通过文件缓存来获取运营商信息，用于算子的初始化。key 为运营商中文名。<br>
   *
   * @param ispInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def ispInfoFetchFromFile(ispInfo: mutable.Map[String, IspEntity]): Unit = {
    LOGGER.info("开始从文件缓存中获取省份信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/ispInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: AnyRef, u: AnyRef) => {
      ispInfo.put(t.toString, JSONUtil.toBean(u.toString, classOf[IspEntity]))
    })
    LOGGER.info(s"从文件缓存中获取运营商信息完成，总共获取到${ispInfo.size}条信息")
  }


  //***************************************************************************获取所有用户信息***************************************************************************

  /**
   * 获取所有用户信息对应的sql语句
   */
  private val userInfoFetchSql: String = "select name, id from cache_log.dim_user order by id limit ? offset ?;"

  /**
   * 从starRocks表中查询所有用户信息。key 为用户名。如果是用于算子初始化，建议使用[[userInfoFetchFromFile]]。
   *
   * @param userInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def userInfoFetch(userInfo: mutable.Map[String, Int]): Unit = {
    LOGGER.info("开始获取所有用户信息")
    val userInfoInner: mutable.Map[String, Int] = mutable.Map[String, Int]()
    val connection: Connection = getConnection
    val prepareStatement: PreparedStatement = connection.prepareStatement(userInfoFetchSql)
    prepareStatement.setInt(1, 10000)
    prepareStatement.setInt(2, 0)
    var selectTimes: Int = 1
    var total: Int = 0
    try {
      userInfoInner.clear()
      breakable(
        //如果select查询获取到的数据量为10000条，则说明表中可能还有数据，则需要再次获取，直到获取到的数量小于10000条。
        while (true) {
          val rs: ResultSet = prepareStatement.executeQuery()
          while (rs.next()) {
            userInfoInner.put(rs.getString("name"), rs.getInt("id"))
            total += 1
          }
          rs.close()
          if (total % 10000 == 0) {
            prepareStatement.setInt(2, 10000 * selectTimes)
            selectTimes += 1
          } else if (total % 10000 != 0) {
            break()
          }
        }
      )
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (userInfoInner.isEmpty) {
      LOGGER.error(s"获取所有用户信息失败，查询信息为：${prepareStatement.toString}")
    } else {
      userInfo.clear()
      userInfoInner.foreach(tuple => userInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有用户信息成功，表数据量：${userInfo.size}")
    }
    prepareStatement.close()
    closeConnection(connection)
  }

  /**
   * 通过文件缓存来获取用户信息，用于算子的初始化。key 为用户名。
   *
   * @param userInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def userInfoFetchFromFile(userInfo: mutable.Map[String, Int]): Unit = {
    LOGGER.info("开始从文件缓存中获取用户信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/userInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: AnyRef, u: AnyRef) => {
      userInfo.put(t.toString, u.toString.toInt)
    })
    LOGGER.info(s"从文件缓存中获取用户信息完成，总共获取到${userInfo.size}条信息")
  }




  //***************************************************************************HTTP***************************************************************************

  //******************************************************************获取所有主机对应节点和cache组信息***************************************************************************
  private val nodeInfoFetchHttpUrl = "http://re-api.bs58i.baishancdnx.com/api"
  private val nodeInfoFetchHttpBody = "{\"method\": \"node.get\",\"token\":\"3243d14fc09936b054f6497dc711a89f\",\"params\":{\"select\":[\"name\",\"ename\",\"cachegroups.svcs.svr.name\",\"cachegroups.btypes.display_value\"]}}"

  /**
   * 通过访问HTTP接口来获取主机对应节点和cache组信息<br>
   * 注意：<b>该函数不可用于算子的初始化，否则会引起高并发问题。算子初始化应该使用[[nodeInfoFetchFromFile]]函数。
   *
   * @param nodeInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def nodeInfoFetch(nodeInfo: mutable.Map[String, NodeEntity]): Unit = {
    val nodeInfoInner: mutable.Map[String, NodeEntity] = mutable.Map[String, NodeEntity]()
    try {
      val httpResult: String = HttpUtil.post(nodeInfoFetchHttpUrl, nodeInfoFetchHttpBody)
      val jSONObject: JSONObject = JSONUtil.parseObj(httpResult)
      if (jSONObject.containsKey("errno") && jSONObject.getInt("errno") == 0 && jSONObject.containsKey("data")) {
        //以下是对json数据的解析，参考原始的数据格式
        jSONObject.getJSONArray("data").jsonIter().forEach(datum => {
          val nodeCode: String = datum.getStr("ename")
          datum.getJSONArray("cachegroups").jsonIter().forEach(cacheGroup => {
            val cacheGroupCode: String = cacheGroup.getStr("ename")
            cacheGroup.getJSONArray("svcs").jsonIter().forEach(svc => {
              val hostname: String = svc.getJSONObject("svr").getStr("name")
              nodeInfoInner.put(hostname, NodeEntity(nodeCode, cacheGroupCode))
            })
          })
        })
      }
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (nodeInfoInner.isEmpty) {
      LOGGER.error(s"获取所有主机对应节点和cache组信息失败")
    } else {
      nodeInfo.clear()
      nodeInfoInner.foreach(tuple => nodeInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有主机对应节点和cache组信息成功，表数据量：${nodeInfo.size}")
    }
  }

  /**
   * 通过本地文件缓存来获取主机对应节点和cache组信息，可用于算子的初始化
   *
   * @param nodeInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def nodeInfoFetchFromFile(nodeInfo: mutable.Map[String, NodeEntity]): Unit = {
    LOGGER.info("开始从文件缓存中获取主机对应节点和cache组信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/nodeInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: scala.Any, u: scala.Any) => {
      nodeInfo.put(t.toString, JSONUtil.toBean(u.toString, classOf[NodeEntity]))
    })
    LOGGER.info(s"从文件缓存中获取主机对应节点和cache组信息完成，总共获取到${nodeInfo.size}条信息")
  }


  //******************************************************************获取所有主机对应typeModule信息***************************************************************************
  private val typeModuleInfoFetchHttpUrl = "https://taishan-api-out.bs58i.baishancloud.com/svr/list?apikey=Lkz9NfE9y2tqfAqAdD83oFBcouRHhZ2m"

  /**
   * 通过访问HTTP接口来获取主机对应typeModule信息<br>
   * 注意：<b>该函数不可用于算子的初始化，否则会引起高并发问题。算子初始化应该使用[[typeModuleInfoFetchFromFile]]函数。
   *
   * @param typeModuleInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def typeModuleInfoFetch(typeModuleInfo: mutable.Map[String, mutable.Set[Int]]): Unit = {
    val typeModuleInfoInner: mutable.Map[String, mutable.Set[Int]] = mutable.Map[String, mutable.Set[Int]]()
    try {
      val httpResult: String = HttpUtil.post(typeModuleInfoFetchHttpUrl, "{}")
      val jSONObject: JSONObject = JSONUtil.parseObj(httpResult)
      if (jSONObject.containsKey("code") && jSONObject.getInt("code") == 0 && jSONObject.containsKey("data")) {
        //以下是对json数据的解析，参考原始的数据格式
        jSONObject.getJSONArray("data").jsonIter().forEach(datum => {
          val hostname: String = datum.getStr("name")
          val typeModules: mutable.Set[Int] = mutable.Set[Int]()
          typeModuleInfoInner.put(hostname, typeModules)
          datum.getJSONArray("svcs").jsonIter().forEach(svc => {
            typeModules.add(svc.getInt("type_module"))
          })
        })
      }
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (typeModuleInfoInner.isEmpty) {
      LOGGER.error(s"获取所有主机对应typeModule信息失败")
    } else {
      typeModuleInfo.clear()
      typeModuleInfoInner.foreach(tuple => typeModuleInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有主机对应typeModule信息成功，表数据量：${typeModuleInfo.size}")
    }
  }

  /**
   * 通过文件缓存来获取主机对应typeModule信息
   *
   * @param typeModuleInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def typeModuleInfoFetchFromFile(typeModuleInfo: mutable.Map[String, mutable.Set[Int]]): Unit = {
    LOGGER.info("开始从文件缓存中获取主机对应typeModule信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/typeModuleInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: scala.Any, u: scala.Any) => {
      val typeModuleSet: mutable.Set[Int] = mutable.Set[Int]()
      JSONUtil.parseArray(u.toString).forEach(element => {
        typeModuleSet.add(element.toString.toInt)
      })
      typeModuleInfo.put(t.toString, typeModuleSet)
    })
    LOGGER.info(s"从文件缓存中获取主机对应typeModule信息完成，总共获取到${typeModuleInfo.size}条信息")
  }


  //******************************************************************获取所有主机对应服务类型信息***************************************************************************

  /**
   * 获取所有主机对应服务类型信息<br>
   * 注意：<b>该函数不可用于算子的初始化，否则会引起高并发问题。算子初始化应该使用[[serviceTypeInfoFetchFromFile]]函数。
   *
   * @param serviceTypeInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def serviceTypeInfoFetch(serviceTypeInfo: mutable.Map[String, Byte]): Unit = {
    val serviceTypeInfoInner: mutable.Map[String, Byte] = mutable.Map[String, Byte]()
    val typeModuleInfo: mutable.Map[String, mutable.Set[Int]] = mutable.Map[String, mutable.Set[Int]]()
    typeModuleInfoFetch(typeModuleInfo)
    typeModuleInfo.foreach(tuple => {
      var serviceType: Byte = 0
      if (tuple._2.contains(361183) && tuple._2.contains(3613013)) {
        serviceType = 1
      } else if (tuple._2.contains(3613013)) {
        serviceType = 2
      } else if (tuple._2.contains(3613478)) {
        serviceType = 3
      } else {
        serviceType = 0
      }
      serviceTypeInfoInner.put(tuple._1, serviceType)
    })
    if (serviceTypeInfoInner.isEmpty) {
      LOGGER.error(s"获取所有主机对应服务类型信息失败")
    } else {
      serviceTypeInfo.clear()
      serviceTypeInfoInner.foreach(tuple => serviceTypeInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有主机对应服务类型信息成功，表数据量：${serviceTypeInfo.size}")
    }
  }

  /**
   * 通过文件缓存来获取主机对应服务类型信息
   *
   * @param serviceTypeInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def serviceTypeInfoFetchFromFile(serviceTypeInfo: mutable.Map[String, Byte]): Unit = {
    LOGGER.info("开始从文件缓存中获取主机对应服务类型信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/serviceTypeInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: scala.Any, u: scala.Any) => {
      serviceTypeInfo.put(t.toString, u.toString.toByte)
    })
    LOGGER.info(s"从文件缓存中获取主机对应服务类型完成，总共获取到${serviceTypeInfo.size}条信息")
  }


  //******************************************************************获取所有主机对应SVG信息***************************************************************************
  private val svgInfoFetchHttpUrl = "https://oms-api.bs58i.baishancdnx.com/apis/scenter-fog-svg"

  /**
   * 通过访问HTTP接口来获取主机对应SVG信息<br>
   * 注意：<b>该函数不可用于算子的初始化，否则会引起高并发问题。算子初始化应该使用[[svgInfoFetchFromFile]]函数。
   *
   * @param svgInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def svgInfoFetch(svgInfo: mutable.Map[String, String]): Unit = {
    val svgInfoInner: mutable.Map[String, String] = mutable.Map[String, String]()
    try {
      val httpResult: String = HttpUtil.get(svgInfoFetchHttpUrl)
      val jSONObject: JSONObject = JSONUtil.parseObj(httpResult)
      if (jSONObject.containsKey("code") && jSONObject.getInt("code") == 0 && jSONObject.containsKey("data")) {
        //以下是对json数据的解析，参考原始的数据格式
        jSONObject.getJSONArray("data").jsonIter().forEach(datum => {
          val svgName: String = datum.getStr("ename")
          val svcs: JSONArray = datum.getJSONArray("svcs")
          if (svcs != null) {
            svcs.jsonIter().forEach((svc: JSONObject) => {
              val hostname: String = svc.getJSONObject("svr").getStr("name")
              svgInfoInner.put(hostname, svgName)
            })
          }
        })
      }
    } catch {
      case e: Exception => LOGGER.error(ExceptionUtils.stringifyException(e))
    }
    if (svgInfoInner.isEmpty) {
      LOGGER.error(s"获取所有主机对应SVG信息失败")
    } else {
      svgInfo.clear()
      svgInfoInner.foreach(tuple => svgInfo.put(tuple._1, tuple._2))
      LOGGER.info(s"获取所有主机对应SVG信息成功，表数据量：${svgInfo.size}")
    }
  }

  /**
   * 通过文件缓存来获取主机对应SVG信息，可用于算子的初始化
   *
   * @param svgInfo 外部缓存map表，函数会将结果写出到该map表中
   */
  def svgInfoFetchFromFile(svgInfo: mutable.Map[String, String]): Unit = {
    LOGGER.info("开始从文件缓存中获取主机对应SVG信息")
    val properties = new Properties()
    properties.load(new InputStreamReader(ExternalDataFetchUtil.getClass.getResourceAsStream("/cache/svgInfo"), CharsetUtil.UTF_8))
    properties.forEach((t: scala.Any, u: scala.Any) => {
      svgInfo.put(t.toString, u.toString)
    })
    LOGGER.info(s"从文件缓存中获取主机对应SVG信息完成，总共获取到${svgInfo.size}条信息")
  }


}









